#include <fstream>
#include <random>
#include <thread>

#include <bthread/bthread.h>
#include <glog/logging.h>
#include <gtest/gtest.h>

#include "hawking/indexlib/index_config.pb.h"
#include "hawking/indexlib/indexlib_data.pb.h"
#include "hawking/util/node_content.pb.h"
#include "doc/document_creator.h"
#include "table/table.h"
#include "util/file_wrapper/file_wrapper.h"

using namespace hawking::indexlib;

std::shared_ptr<IndexlibIndexConfig> MakeIndexConfig() {
    std::shared_ptr<IndexlibIndexConfig> index_config = 
        std::make_shared<IndexlibIndexConfig>();
    int32_t field_id = 0;
    // i8
    auto* field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i8");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT8);
    // i32
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i32");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT32);
    // i64
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i64");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT64);
    // float
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("float");
    field->set_field_type(hawking::util::ColumnType::COLUMN_FLOAT);
    // double
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("double");
    field->set_field_type(hawking::util::ColumnType::COLUMN_DOUBLE);
    // string
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("string");
    field->set_field_type(hawking::util::ColumnType::COLUMN_STRING);
    // i8s
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i8s");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT8_LIST);
    // i32s
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i32s");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT32_LIST);
    // i64s
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("i64s");
    field->set_field_type(hawking::util::ColumnType::COLUMN_INT64_LIST);
    // floats
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("floats");
    field->set_field_type(hawking::util::ColumnType::COLUMN_FLOAT_LIST);
    // doubles
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("doubles");
    field->set_field_type(hawking::util::ColumnType::COLUMN_DOUBLE_LIST);
    // strings
    field = index_config->add_fields();
    field->set_field_id(field_id++);
    field->set_field_name("strings");
    field->set_field_type(hawking::util::ColumnType::COLUMN_STRING_LIST);

    // inv1
    auto* inverted_configs = index_config->add_inverted_configs();
    inverted_configs->set_field_id(1);
    inverted_configs->set_field_name("i32");

    // inv2
    inverted_configs = index_config->add_inverted_configs();
    inverted_configs->set_field_id(5);
    inverted_configs->set_field_name("string");

    // pk
    index_config->set_pk_field_id(2);

    return index_config;
}

void MakeMsg(IndexlibMessage* msg, int64_t fake) {
    auto* field = msg->add_columns();
    field->set_column_name("i8");
    field->mutable_column_value()->set_int_value(static_cast<int8_t>(fake));

    field = msg->add_columns();
    field->set_column_name("i32");
    field->mutable_column_value()->set_int_value(static_cast<int32_t>(fake + 100000));

    field = msg->add_columns();
    field->set_column_name("i64");
    field->mutable_column_value()->set_int_value(fake + 10000000);

    field = msg->add_columns();
    field->set_column_name("float");
    field->mutable_column_value()->set_double_value(static_cast<float>(fake/100000));

    field = msg->add_columns();
    field->set_column_name("double");
    field->mutable_column_value()->set_double_value(static_cast<double>(fake/200000));

    field = msg->add_columns();
    field->set_column_name("string");
    field->mutable_column_value()->set_bytes_value(std::to_string(fake));

    field = msg->add_columns();
    field->set_column_name("i8s");
    field->mutable_column_value()->mutable_int_values()->add_values(static_cast<int8_t>(fake));
    field->mutable_column_value()->mutable_int_values()->add_values(static_cast<int8_t>(fake + 1));
    field->mutable_column_value()->mutable_int_values()->add_values(static_cast<int8_t>(fake + 2));

    field = msg->add_columns();
    field->set_column_name("i32s");
    field->mutable_column_value()->mutable_int_values()->add_values(static_cast<int32_t>(fake + 20000));
    field->mutable_column_value()->mutable_int_values()->add_values(static_cast<int32_t>(fake + 80000));
    field->mutable_column_value()->mutable_int_values()->add_values(static_cast<int32_t>(fake + 100000));

    field = msg->add_columns();
    field->set_column_name("i64s");
    field->mutable_column_value()->mutable_int_values()->add_values(fake + 100000001);
    field->mutable_column_value()->mutable_int_values()->add_values(fake + 100000002);
    field->mutable_column_value()->mutable_int_values()->add_values(fake + 100000003);

    field = msg->add_columns();
    field->set_column_name("floats");
    field->mutable_column_value()->mutable_double_values()->add_values(static_cast<float>(fake + 1000));
    field->mutable_column_value()->mutable_double_values()->add_values(static_cast<float>(fake + 2000));
    field->mutable_column_value()->mutable_double_values()->add_values(static_cast<float>(fake + 5000));

    field = msg->add_columns();
    field->set_column_name("doubles");
    field->mutable_column_value()->mutable_double_values()->add_values(static_cast<double>(fake + 150));
    field->mutable_column_value()->mutable_double_values()->add_values(static_cast<double>(fake + 250));
    field->mutable_column_value()->mutable_double_values()->add_values(static_cast<double>(fake + 550));

    field = msg->add_columns();
    field->set_column_name("strings");
    field->mutable_column_value()->mutable_bytes_values()->add_values(std::to_string(fake + 10));
    field->mutable_column_value()->mutable_bytes_values()->add_values(std::to_string(fake + 50));
    field->mutable_column_value()->mutable_bytes_values()->add_values(std::to_string(fake + 100));
}

void MakeAddMsg(IndexlibMessage* msg, int64_t fake) {
    MakeMsg(msg, fake);
    msg->set_cmd(hawking::util::IndexOp::INDEX_RT_ADD);
}

void MakeUpdateMsg(IndexlibMessage* msg, int64_t fake) {
    MakeMsg(msg, fake);
    msg->set_cmd(hawking::util::IndexOp::INDEX_RT_UPDATE);
}

void MakeDelMsg(IndexlibMessage* msg, int64_t fake) {
    MakeMsg(msg, fake);
    msg->set_cmd(hawking::util::IndexOp::INDEX_RT_DEL);
}

TEST(table_test, localdata_filter) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    Table table("task1", "/tmp/test_table");
    EXPECT_EQ(true, table.Init(index_config));

    // add
    for (int64_t fake = 0; fake < 102400; ++fake) {
        IndexlibMessage message;
        MakeAddMsg(&message, fake);
        EXPECT_EQ(true, table.AddDocument(message));
    }

    // dump
    EXPECT_EQ(true, table.Save());

    // load
    Table table2("task1", "/tmp/test_table");
    EXPECT_EQ(true, table2.Load(index_config, false, false));

    // filter
    google::protobuf::RepeatedPtrField<hawking::util::FilterColumn> filter_columns;
    google::protobuf::RepeatedPtrField<hawking::util::FilterColumn> where_columns;
    hawking::util::FilterColumn* filter_eq = filter_columns.Add();
    int64_t pk = 10000002;

    filter_eq->set_column_name("i8");
    filter_eq->mutable_column_value()->set_int_value(2);
    filter_eq->set_action(hawking::util::ExpressionOp::EQ);
    auto ret = table2.FilterCheck(pk, filter_columns, where_columns);
    EXPECT_EQ(ret, true);
}

TEST(table_test, localdata_select) {
    std::shared_ptr<IndexlibIndexConfig> index_config = MakeIndexConfig();
    Table table("task1", "/tmp/test_table");
    EXPECT_EQ(true, table.Init(index_config));

    // add
    for (int64_t fake = 0; fake < 102400; ++fake) {
        IndexlibMessage message;
        MakeAddMsg(&message, fake);
        EXPECT_EQ(true, table.AddDocument(message));
    }

    // dump
    EXPECT_EQ(true, table.Save());

    // load
    Table table2("task1", "/tmp/test_table");
    EXPECT_EQ(true, table2.Load(index_config, false, false));

    // select
    google::protobuf::RepeatedPtrField<hawking::util::SelectColumn> select_columns;
    hawking::util::SelectColumn* col_i8 = select_columns.Add();
    col_i8->set_column_name("i8");
    hawking::util::SelectColumn* col_i32 = select_columns.Add();
    col_i32->set_column_name("i32");
    hawking::util::SelectColumn* col_i64 = select_columns.Add();
    col_i64->set_column_name("i64");
    hawking::util::SelectColumn* col_float = select_columns.Add();
    col_float->set_column_name("float");
    hawking::util::SelectColumn* col_double = select_columns.Add();
    col_double->set_column_name("double");
    hawking::util::SelectColumn* col_string = select_columns.Add();
    col_string->set_column_name("string");
    hawking::util::SelectColumn* col_i8s = select_columns.Add();
    col_i8s->set_column_name("i8s");
    hawking::util::SelectColumn* col_i32s = select_columns.Add();
    col_i32s->set_column_name("i32s");
    hawking::util::SelectColumn* col_i64s = select_columns.Add();
    col_i64s->set_column_name("i64s");
    hawking::util::SelectColumn* col_floats = select_columns.Add();
    col_floats->set_column_name("floats");
    hawking::util::SelectColumn* col_doubles = select_columns.Add();
    col_doubles->set_column_name("doubles");
    hawking::util::SelectColumn* col_strings = select_columns.Add();
    col_strings->set_column_name("strings");
    int64_t pk = 10000002;
    table2.FillLocaldata(pk, &select_columns);
    for (const auto& col : select_columns) {
        LOG(INFO) << col.ShortDebugString();
    }
}

int main(int argc, char* argv[]) {
    ::testing::InitGoogleTest(&argc, argv);
    ::google::ParseCommandLineFlags(&argc, &argv, true);
    srand(static_cast<unsigned>(time(0)));
    return RUN_ALL_TESTS();
}